package com.shujia.spark

import org.apache.spark.broadcast.Broadcast
import org.apache.spark.rdd.RDD
import org.apache.spark.{SparkConf, SparkContext}

object Demo18Bro {

  def main(args: Array[String]): Unit = {
    val conf: SparkConf = new SparkConf()
      .setAppName("pi")
      .setMaster("local")

    val sc = new SparkContext(conf)


    val studentRDD: RDD[String] = sc.textFile("spark/data/students.txt")


    /**
      * RDD 不能嵌套使用
      * 在算子中不能在使用rdd
      *
      */
    //    studentRDD.foreach(line => {
    //      studentRDD.foreach(println)
    //    })

    /**
      * 在算子中不能修改算子外面的变量,修改改了在算子外面也不会生效
      *
      */


    var i = 0

    studentRDD.foreach(line => {
      i += 1
      println(i)
      println(line)
    })

    println(i)


    //假设集合元素较多
    /*val clazzs = List("文科一班", "文科二班", "文科三班", "理科一班")

    val filterRDD: RDD[String] = studentRDD.filter(line => {

      val clazz: String = line.split(",")(4)

      //在算子里面使用的Driver端的变量,
      //每一个task中都会封装一个变量副本
      clazzs.contains(clazz)
    })

    filterRDD.foreach(println)*/


    /**
      *
      * 广播变量
      *
      * 只能在Driver端定义
      * 只能在Executor端读取
      *
      */


    //普通变量
    val clazzs = List("文科一班", "文科二班", "文科三班", "理科一班")

    //将scala集合广播出去
    //广播变量
    val broClazz: Broadcast[List[String]] = sc.broadcast(clazzs)


    val filterRDD: RDD[String] = studentRDD.filter(line => {
      val clazz: String = line.split(",")(4)

      //在算子中获取广播变量
      val value: List[String] = broClazz.value

      value.contains(clazz)
    })

    filterRDD.foreach(println)


  }

}
